package org.zjt.flink;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

/**
 * Description:   参数传递 Parameter
 *
 * @author juntao.zhang
 * Date: 2018-10-09 下午6:06
 * @see
 */
public class ParametersJob {

    private static final String PARAM_NAME = "limit";

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Integer> toFilter = env.fromElements(1, 2, 3);

        Configuration config = new Configuration();
        config.setInteger(PARAM_NAME, 2);
        toFilter.filter(new RichFilterFunction<Integer>() {
            private int limit;

            @Override
            public void open(Configuration parameters) throws Exception {
                limit = parameters.getInteger(PARAM_NAME, 0);
            }

            @Override
            public boolean filter(Integer value) throws Exception {
                return value > limit;
            }
        }).withParameters(config).print();
    }
}
